package Autodesk;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.Hashtable;
import java.util.List;
import java.util.Map;
import java.util.StringTokenizer;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.filecache.DistributedCache;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.Reducer.Context;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;
import org.apache.log4j.Logger;

import Autodesk.ParseLogs.LogMapper;
import Autodesk.ParseLogs.LogReducer;

public class HaProxyLogsParser  extends Configured implements Tool {
	private static Logger logger = Logger.getLogger(ParseLogs.class);
	public Path prodPath;
	public static class LogMapper extends Mapper<Object, Text, Text, IntWritable> {
		
		private static final String EXCLUDE_VALUES = "beehive_monitor";  
		private final static IntWritable one = new IntWritable(1);
		private Text word = new Text();
		private Map<ArrayList<String>,String> StaticListLang=new HashMap<ArrayList<String>,String>();
		private Map<ArrayList<String>,String> StaticListProd=new HashMap<ArrayList<String>,String>();
		private Map<ArrayList<String>,String> StaticListChannel=new HashMap<ArrayList<String>,String>();
		private Map<ArrayList<String>,String> StaticListCountry=new HashMap<ArrayList<String>,String>();
		private Map<ArrayList<String>,String> StaticListIp=new HashMap<ArrayList<String>,String>();
		private ArrayList<String> listChannelsExcluded=new ArrayList<String>();
//		private Map<ArrayList<String>,String> StaticListFeature=new HashMap<ArrayList<String>,String>();
		Parser par=new Parser();
		

		public void setup(Context context){
			
			Configuration conf=context.getConfiguration();
			 if (conf.getBoolean("ip.patterns",false )){
				  Path[] patternsFiles = new Path[0];
				  try{
			           patternsFiles = DistributedCache.getLocalCacheFiles(conf);
				  }catch(IOException ioe){
			           System.err.println("Caught exception while getting cached files: ");
				  }
				  for (Path patternsFile : patternsFiles) {
					  par.getStaticList(patternsFile,StaticListIp,"ip"); 
		 				logger.info("Loaded Ips-------------------");
				  }
			 }
	 				StaticListLang=par.getStaticList("language"); 
	 				logger.info("Loaded Langs-------------------");
			  		
			  		StaticListProd=par.getStaticList("product"); 
			  		logger.info("Loaded Products-------------------");
			  		
			  		StaticListChannel=par.getStaticList("channel"); 
			  		logger.info("Loaded Channels-------------------");
			  		
			  		StaticListCountry=par.getStaticList("country");
	 				logger.info("Loaded Countries-------------------");

	 				//				  		StaticListFeatures=par.getStaticList("features");
//			  		System.out.println("Loaded Features-------------------");
			 inicitializeListChannelsExc();

		}


		private void inicitializeListChannelsExc() {
			listChannelsExcluded.add("productchannel");
			 listChannelsExcluded.add("productchannel2");
			 listChannelsExcluded.add("supportchannel");
			 listChannelsExcluded.add("supportchannel2");
			 listChannelsExcluded.add("acadconnectchannel");
			 listChannelsExcluded.add("acadconnectchannel2");
			 listChannelsExcluded.add("taxonomy3");

		}
					
	private String getParameter(String parameter,Hashtable<String, String> ht){
		String result;
		if ((ht.get(parameter) != null) && !(ht.get(parameter).isEmpty()))
			result=ht.get(parameter).toString();
		else
			result="2";
		return result;
	}

	private String getQueryResult(String line){
		String result="0";
		if(line.indexOf("x-query-result=")>-1){
			line = line.substring(line.indexOf("x-query-result=") + 15);
			result = line.substring(0,line.indexOf(" "));
			if(result.equals("")){
				result="0";
			}
		}
		return result;
	}

	private StringBuffer getOutput(StringBuffer bf,Hashtable<String, String> ht,String date,String channel,String ipPort,
			String hour,String source,String queryResult,String country){

		bf.append(date + "\t");//Date
		bf.append(channel.equals(null) ? "2" : channel);//Channel
		bf.append("\t");
		bf.append(ipPort + "\t");//IpPort
		bf.append(getParameter("p",ht));//Products
		bf.append("\t");
		bf.append(getParameter("l",ht));//Languages
		bf.append("\t");
		bf.append(hour);//Hour
		bf.append("\t");
		bf.append(getParameter("q",ht));//Query
		bf.append("\t");
		bf.append(queryResult);//QueryResult
		bf.append("\t");
		bf.append(country);//Country
//		bf.append("\t");
//		bf.append(getParameter("s",ht));
//		bf.append("\t");
//		bf.append(getParameter("aud",ht));
//		bf.append("\t");
//		bf.append(getParameter("cg",ht));
//		bf.append("\t");
//		bf.append(getParameter("origin",ht));
//		bf.append("\t");
//		bf.append(getParameter("sc",ht));
//		bf.append("\t");
//		bf.append(getParameter("c",ht));
		
		return bf;
	}

	
	//MAP FUNCTION
	public void map(Object key, Text value, Context context) throws IOException, InterruptedException {

		String line = value.toString().toLowerCase();
		Hashtable<String, String> ht = new Hashtable<String, String>();
		String date = null;
		String channel = null;
		String source=null;
		ArrayList<String> listIpForw=new ArrayList<String>();
		// Looking for DATE
		date = par.getDateHaProxyLogs(line);
	
		String hour=par.getHour(line);
		
		// Prepare Line
		line = line.toLowerCase();
		
		// Looking for IP.
		String ipPort = par.getIpHaProxy(line);
		if(ipPort.isEmpty()){
			ipPort="2";
		}
		
		// Looking for listForward.
		listIpForw=par.getListIPForwaredFor(line);
		String ipExcluded="1";
		if(!listIpForw.isEmpty()){
			for(int h=0;h<listIpForw.size();h++){
				ipExcluded=par.lookForKey(StaticListIp, listIpForw.get(h), "ip");
			}
		}
		
		// Looking for CHANNEL.
		String channelID;	
		channel = par.getChannel(line);
		channelID= par.lookForKey(StaticListChannel, channel,"channel");
		 //Looking for Country
		String country=par.getCountry(line);
		country=par.lookForKey(StaticListCountry,country,"country");

		//LOOKING FOR QueryResult
		String queryResult="0";//Amount of times that the specified query appears 

		 if( (!channelID.equals("1")) && (ipExcluded.equals("1")) && (line.indexOf(EXCLUDE_VALUES)==-1) && (line.indexOf('?')>-1)){
		 
			 	String queryLine=line.substring(line.indexOf('?')+1);
			 	StringTokenizer tokenizer = new StringTokenizer(queryLine,"&");
				while (tokenizer.hasMoreTokens()) {
					String token = tokenizer.nextToken();
						
						if (token.indexOf("q=") > -1) {
							String query=token.substring(2);
							if(query.indexOf(" http")>-1){
								query=query.substring(0,query.indexOf("http"));
							}
								
							ht.put("q",query) ;
						} else if (token.indexOf("p=") > -1) {
							String id;
							try{
								String prodAb=par.getProduct(token, queryLine,par);								
								String[] productRele=prodAb.split("\t");
								String tokenProd=productRele[0];
								String tokenRele=productRele[1];
								id=par.lookForKeyProduct(StaticListProd, tokenProd.toLowerCase(),tokenRele.toLowerCase());
							}catch(Exception e){
								id="2";
							}
							ht.put("p", id);
						} else if ((token.indexOf("l=") > -1) && (token.indexOf("channel=") < 0) && (token.indexOf("url=") < 0)) {

							String tokenSub = token.substring(2);
							
							String id = par.lookForKey(StaticListLang, tokenSub,"language");
							ht.put("l", id);
							
//						} else if (token.indexOf("source=") > -1) {
//							token = token.substring(token.indexOf("=")+1);
//							String id = par.lookForKey(StaticListFeature, token,"features");
//							ht.put("s", id);
//						}else if (token.indexOf("audience=") > -1) {
//							token = token.substring(token.indexOf("=")+1);
//							String id = par.lookForKey(StaticListFeature, token,"features");
//							ht.put("aud", id);
//						}else if (token.indexOf("sc=") > -1) {
//							token = token.substring(token.indexOf("=")+1);
//							String id = par.lookForKey(StaticListFeature, token,"features");
//							ht.put("sc",id);	
//
//						}else if (token.indexOf("c=") > -1) {
//							token = token.substring(token.indexOf("=")+1);
//							String id = par.lookForKey(StaticListFeature, token,"features");
//							ht.put("c", id);
//
//						}else if (token.indexOf("cg=") > -1) {
//							token = token.substring(token.indexOf("=")+1);
//							String id = par.lookForKey(StaticListFeature, token,"features");
//							ht.put("cg",id);
//						}else if (token.indexOf("origin=") > -1) {
//							token = token.substring(token.indexOf("=")+1);
//							String id = par.lookForKey(StaticListFeature, token,"features");
//							ht.put("origin", id);
						}
					}
			}
			StringBuffer bf = new StringBuffer();
			if (ht != null && !ht.isEmpty() ) {
				bf = getOutput(bf, ht, date, channelID, ipPort,hour,source,queryResult,country);
				word.set(bf.toString());
				context.write(word, one);

			}
	}

}

//REDUCE FUNCTION	
public static class LogReducer extends
		Reducer<Text, IntWritable, Text, IntWritable> {
	
	private IntWritable result = new IntWritable();

	public void reduce(Text key, Iterable<IntWritable> values,
			Context context) throws IOException, InterruptedException {
		
		int sum = 0;
		for (IntWritable val : values) {
			sum += val.get();
		}
		result.set(sum);
		context.write(key, result);
	
	}

}

public int run(String[] args) throws Exception { 
	Configuration conf = getConf();
	
	Job job = new Job(conf, "ParseLogs");
	job.setJarByClass(Autodesk.ParseLogs.class);
	
	
	job.setMapperClass(LogMapper.class);
	job.setReducerClass(LogReducer.class);
	
	job.setOutputKeyClass(Text.class);
	job.setOutputValueClass(IntWritable.class);
	
	List<String> other_args = new ArrayList<String>();
	for (int i = 0; i < args.length; ++i) {
		if ("-ip".equals(args[i])) {
			prodPath=new Path(args[++i]);
			DistributedCache.addCacheFile(prodPath.toUri(),
					job.getConfiguration());
			job.getConfiguration().setBoolean("ip.patterns",
					true);
			job.getConfiguration().set("ip.path", prodPath.getName());
		}else{
			other_args.add(args[i]);
		}
	}

    FileInputFormat.setInputPaths(job, new Path(other_args.get(0)));
    FileOutputFormat.setOutputPath(job, new Path(other_args.get(1)));
    
	System.exit(job.waitForCompletion(true)?0:1);
	
	return 0;

}

public static void main(String[] args) throws Exception {
	int res = ToolRunner.run(new Configuration(), new HaProxyLogsParser(), args);
	System.exit(res);
	}
}
